package com.hydra.auth.service;

import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.IService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hydra.auth.common.SerialNoRuleEnum;
import com.hydra.auth.entity.BasicSerialNo;
import com.hydra.auth.mapper.BasicSerialNoMapper;
import com.hydra.base.redis.RedisService;
import com.hydra.base.utils.JsonUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.temporal.TemporalAdjusters;
import java.util.Objects;

/**
 * <p>
 *  发号器
 *  参考周雁杰油卡逻辑
 * </p>
 *
 * @author liangyongtong
 * @since 2020-03-26
 */
@SuppressWarnings("all")
@Component
public class SerialNoService extends ServiceImpl<BasicSerialNoMapper, BasicSerialNo> implements IService<BasicSerialNo> {

    protected Logger log = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private RedisService redisService;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 前缀标志
     */
    private static final String FLAG_PREFIX = "GENERATE_NO:SYY";

    /**
     * 扩容阈值， 小于则发起扩容，  必须大于并发数 否则可能出现获取失败的情况
     */
    private static final int CAPACITY_THRESHOLD = 500;

    /**
     * 每批次取数， 必须大于程序并发数， 否则可能出现获取失败的情况
     */
    private static final Long BATCH_SIZE = 1000L;

    /**
     * 重试最大次数
     */
    private static final int RETRY_LIMIT = 3;

    /**
     * 重试间隔时间
     */
    private static final long RETRY_TIME = 3000L;

    enum RedisKey {
        /**
         * 当前序号
         */
        CUR_SEQ_NO_PREFIX,

        /**
         * 序号配置
         */
        SEQ_NO_CONF_PREFIX,

        /**
         * 初始化序号
         */
        INIT_SEQ_NO_PREFIX,

        /**
         * 重置序号
         */
        REST_SEQ_NO_PREFIX,

        /**
         * 扩容序号
         */
        CAPACITY_SEQ_NO_PREFIX
    }

    @Transactional(propagation = Propagation.NOT_SUPPORTED)
    public String generateCode(String type) {

        if (StringUtils.isBlank(type)) {
            throw new RuntimeException("序号不能为空");
        }

        //重试次数
        int retryNum = 0;

        //1.获取可用的缓存配置
        BasicSerialNo cacheSerialNo = getCacheAvailableSerialNo(type, retryNum);

        Assert.notNull(cacheSerialNo, "系统错误，没有可用的序号");

        //2.序号++
        String seqKey = getRedisKey(type, RedisKey.CUR_SEQ_NO_PREFIX);

        Long seqNo = redisService.incr(seqKey);

        //3.校验序号
        checkSeq(seqNo, cacheSerialNo, type);

        //4.扩容判断
        checkCapacity(seqNo, cacheSerialNo, type);

        //5.拼接序号返回
        return aggregateResult(seqNo, cacheSerialNo);
    }


    private String aggregateResult(Long seqNo, BasicSerialNo cacheSerialNo) {
        StringBuilder result = new StringBuilder();
        //拼接前缀
        result.append(cacheSerialNo.getSeqPrefix());
        //拼接日期
        result.append(formatPattern(cacheSerialNo.getLatestDate(), cacheSerialNo.getFormatPattern()));
        //拼接序号
        result.append(StringUtils.leftPad(seqNo.toString(), cacheSerialNo.getSeqLpadLength(), "0"));

        String resultStr = result.toString();
        log.info("generateCode:{}", resultStr);
        return resultStr;
    }

    /**
     * %Y
     * %m
     * %d
     *
     * @param pattern
     * @return
     */
    private String formatPattern(LocalDate localDate, String pattern) {
        if (StringUtils.isNotBlank(pattern)) {
            // 没有日期，设置当前日期
            if (Objects.isNull(localDate)) {
                localDate = LocalDate.now();
            }
            return pattern.replace("%Y", String.valueOf(localDate.getYear())).
                    replace("%m", StringUtils.leftPad(String.valueOf(localDate.getMonthValue()), 2, "0")).
                    replace("%d", StringUtils.leftPad(String.valueOf(localDate.getDayOfMonth()), 2, "0"));
        } else {
            return StringUtils.EMPTY;
        }
    }


    /**
     * 扩容校验
     *
     * @param seqNo
     * @param cacheSerialNo
     * @param type
     */
    private void checkCapacity(Long seqNo, BasicSerialNo cacheSerialNo, String type) {
        long size = cacheSerialNo.getSeqNo() - seqNo;
        log.info("[{}]:缓存上限:[{}],剩余数:[{}]", type, cacheSerialNo.getSeqNo(), size);
        //如果剩余号小于阈值，则发起扩容
        if (size < CAPACITY_THRESHOLD) {
            BasicSerialNo BasicSerialNo = this.getById(type);
            Assert.notNull(BasicSerialNo, "获取序号配置失败，请检查数据库配置：" + type);
            capacitySerialNo(BasicSerialNo, type);
        }
    }

    /**
     * 扩容序号
     *
     * @param type
     */
    private BasicSerialNo capacitySerialNo(BasicSerialNo BasicSerialNo, String type) {
        String lockKey = getRedisKey(type, RedisKey.CAPACITY_SEQ_NO_PREFIX);
        //1.加锁, 15秒， 不重试
        boolean tryLock = redisService.lockByNXPX(lockKey, 1500L);
        //如果拿到锁则进行扩容，否则跳过
        if (tryLock) {
            log.info("开始扩容");
            try {
                Long oilVersion = BasicSerialNo.getVersion();
                Long originSeqNo = BasicSerialNo.getSeqNo() == null ? 0 : BasicSerialNo.getSeqNo();
                //新的
                Long newSeqNo = originSeqNo + BATCH_SIZE;
                String seqVal = aggregateResult(newSeqNo, BasicSerialNo);

                BasicSerialNo updateNo = new BasicSerialNo();

                updateNo.setSeqNo(newSeqNo);
                updateNo.setSeqVal(seqVal);
                updateNo.setVersion(oilVersion + 1);
                updateNo.setLatestDate(LocalDate.now());

                UpdateWrapper updateWrapper = new UpdateWrapper();
                updateWrapper.eq("SEQ_CODE", type);
                updateWrapper.eq("version", oilVersion);
                boolean result = this.update(updateNo, updateWrapper);

//                int count = serialNoDao.updateByVersion(newSeqNo, seqVal, oilVersion, LocalDate.now(), type);
                //如果修改成功
                if (result) {
                    redisTemplate.executePipelined(new RedisCallback<Object>() {
                        @Override
                        public Object doInRedis(RedisConnection connection) throws DataAccessException {

                            byte[] noKey = getRedisKey(type, RedisKey.CUR_SEQ_NO_PREFIX).getBytes();
                            byte[] noVal = originSeqNo.toString().getBytes();

                            //key 不存在时才设置, 默认为
                            connection.setNX(noKey, noVal);

                            //更新缓存配置
                            BasicSerialNo.setSeqNo(newSeqNo);
                            BasicSerialNo.setSeqVal(seqVal);
                            BasicSerialNo.setVersion(oilVersion + 1);
                            BasicSerialNo.setLatestDate(LocalDate.now());
                            setRedisConf(connection, BasicSerialNo, type);
                            return null;
                        }
                    });

                    log.info("[{}]:缓存上限：[{}], 扩容完毕", BasicSerialNo.getSeqCode(), BasicSerialNo.getSeqNo());
                    return BasicSerialNo;
                } else {
                    log.error("扩容修改序号失败");
                }
            } finally {
                redisService.releaseLockByNXPX(lockKey);
            }
        } else {
            log.info("获锁失败，跳过扩容");
        }
        return null;
    }


    /**
     * 校验序号
     *
     * @param seq
     * @param BasicSerialNo
     */
    private void checkSeq(Long seq, BasicSerialNo BasicSerialNo, String type) {
        try {
            Assert.notNull(seq, "生成序号为null");
            Assert.isTrue(seq.longValue() > 0, "生成序号为负数");
            Assert.isTrue(seq.longValue() <= BasicSerialNo.getSeqNo().longValue(), "生成序号超过上限");
            Assert.isTrue(seq.toString().length() <= BasicSerialNo.getSeqLpadLength().intValue(), "生成序号超过限制长度");
        } catch (Exception e) {
            clearCacheSerialNo(type);
            throw e;
        }
    }

    /**
     * 初始化序号
     *
     * @param type
     * @return
     */
    private BasicSerialNo initSerialNo(String type, int retryNum) {
        log.info("初始化");
        //1.锁住
        String lockKey = getRedisKey(type, RedisKey.INIT_SEQ_NO_PREFIX);
        //锁住15秒， 不重试
        boolean tryLock = redisService.lockByNXPX(lockKey, 15000L);
        if (tryLock) {
            try {
                //查询出配置
                BasicSerialNo basicSerialNo = this.getOne(Wrappers.<BasicSerialNo>lambdaQuery().eq(BasicSerialNo::getSeqCode, type).last(" for update"));
                Assert.notNull(basicSerialNo, "获取序号配置失败，请检查数据库配置：" + type);
                //看是否需要重置
                BasicSerialNo check = checkRestGet(basicSerialNo, type, retryNum);
                if (check != null) {
                    //如果重置成功则直接返回
                    return check;
                } else {
                    //进行初始化
                    return capacitySerialNo(basicSerialNo, type);
                }
            } finally {
                redisService.releaseLockByNXPX(lockKey);
            }
        } else {
            //没有获取到锁，则重试
            return retry(type, retryNum);
        }
    }

    /**
     * 清理缓存的序号配置
     *
     * @param type
     */
    private void clearCacheSerialNo(String type) {
        String prefixKey = getRedisKey(type, RedisKey.SEQ_NO_CONF_PREFIX);
        String noPrefix = getRedisKey(type, RedisKey.CUR_SEQ_NO_PREFIX);

        redisService.deleteKeys(prefixKey, noPrefix);
    }

    /**
     * 重置序号
     *
     * @param type
     * @return
     */
    private BasicSerialNo resetSerialNo(String type, int retryNum) {
        log.info("重置序号");
        //1.锁住
        String lockKey = getRedisKey(type, RedisKey.REST_SEQ_NO_PREFIX);
        //锁住15秒， 不重试
        boolean tryLock = redisService.lockByNXPX(lockKey, 15000L);
        if (tryLock) {
            //如果获取到锁
            try {
                //查询出配置
                BasicSerialNo BasicSerialNo = this.getById(type);
                Assert.notNull(BasicSerialNo, "获取序号配置失败，请检查数据库配置：" + type);
                //原来第版本号
                Long oilVersion = BasicSerialNo.getVersion();
                String seqVal = aggregateResult(BATCH_SIZE, BasicSerialNo);

                BasicSerialNo updateNo = new BasicSerialNo();

                updateNo.setSeqNo(BATCH_SIZE);
                updateNo.setSeqVal(seqVal);
                updateNo.setVersion(oilVersion + 1);
                updateNo.setLatestDate(LocalDate.now());

                UpdateWrapper updateWrapper = new UpdateWrapper();
                updateWrapper.eq("SEQ_CODE", type);
                updateWrapper.eq("version", oilVersion);
                boolean result = this.update(updateNo, updateWrapper);

                if (result) {

                    redisTemplate.executePipelined(new RedisCallback<Object>() {
                        @Override
                        public Object doInRedis(RedisConnection connection) throws DataAccessException {
                            byte[] noKey = getRedisKey(type, RedisKey.CUR_SEQ_NO_PREFIX).getBytes();
                            byte[] noVal = "0".getBytes();
                            //覆盖当前值
                            connection.set(noKey, noVal);

                            //更新缓存
                            BasicSerialNo.setSeqNo(BATCH_SIZE);
                            BasicSerialNo.setSeqVal(seqVal);
                            BasicSerialNo.setVersion(oilVersion + 1);
                            BasicSerialNo.setLatestDate(LocalDate.now());
                            setRedisConf(connection, BasicSerialNo, type);
                            return null;
                        }
                    });

                    return BasicSerialNo;
                } else {
                    log.info("重置序号更新数据库失败");
                }
            } finally {
                redisService.releaseLockByNXPX(lockKey);
            }
        } else {
            //没有获取到锁，则重试
            return retry(type, retryNum);
        }
        return null;
    }

    private void setRedisConf(RedisConnection connection, BasicSerialNo BasicSerialNo, String type) {
        byte[] conf = JsonUtil.toJsonString(BasicSerialNo).getBytes();
        Integer expire = getExpireSeconds(BasicSerialNo.getSeqNoResetRule());
        if (expire != null) {
            log.info(" rule:{},  expire:{}", BasicSerialNo.getSeqNoResetRule(), expire);
            connection.setEx(getRedisKey(type, RedisKey.SEQ_NO_CONF_PREFIX).getBytes(), expire, conf);
        } else {
            connection.set(getRedisKey(type, RedisKey.SEQ_NO_CONF_PREFIX).getBytes(), conf);
        }
    }

    /**
     * 获取截止过期时间， 秒
     *
     * @param restRuleStr
     * @return
     */
    private Integer getExpireSeconds(String restRuleStr) {
        SerialNoRuleEnum restRule = SerialNoRuleEnum.getByName(restRuleStr);
        if (restRule != null) {
            LocalDateTime now = LocalDateTime.now();
            //当天 23：59：59
            LocalDateTime lastDayTime = LocalDateTime.of(LocalDate.now(), LocalTime.MAX);
            LocalDateTime end = null;
            switch (restRule) {
                case DAY:
                    end = LocalDateTime.of(LocalDate.now(), LocalTime.MAX);
                    break;
                case MONTH:
                    end = lastDayTime.with(TemporalAdjusters.lastDayOfMonth());
                    break;
                case YEAR:
                    end = lastDayTime.with(TemporalAdjusters.lastDayOfYear());
                    break;
                default:
                    return null;
            }
            return Integer.valueOf(Duration.between(now, end).toMillis() / 1000 + "");
        }
        return null;
    }


    /**
     * 重试
     *
     * @param type
     * @return
     */
    private BasicSerialNo retry(String type, int retryNum) {
        //如果没有获取到锁
        try {
            //休眠五秒 后再次获取
            log.info("进入休眠等待重试");
            Thread.sleep(RETRY_TIME);
            //如果重试次数小于最大阈值，重试获取
            if (retryNum < RETRY_LIMIT) {
                return getCacheAvailableSerialNo(type, retryNum);
            } else {
                log.info("超过最大重试次数");
            }
        } catch (InterruptedException e) {
            log.error("休眠出现异常", e);
        }
        return null;
    }

    /**
     * 获取可用的缓存序号
     *
     * @param type
     * @return
     */
    private BasicSerialNo getCacheAvailableSerialNo(String type, int retryNum) {
        retryNum++;
        log.info("第{}次获取配置", retryNum);
        String confKey = getRedisKey(type, RedisKey.SEQ_NO_CONF_PREFIX);
        String jsonConfStr = redisService.get(confKey);
        if (StringUtils.isBlank(jsonConfStr)) {
            //如果配置为空, 则初始化
            return initSerialNo(type, retryNum);
        } else {
            try {
                BasicSerialNo BasicSerialNo = JsonUtil.parseObject(jsonConfStr, BasicSerialNo.class);
                //查看是否需要重置,
                BasicSerialNo check = checkRestGet(BasicSerialNo, type, retryNum);
                //如果重置了则返回重置的
                if (check != null) {
                    return check;
                }
                //不需要重置，直接返回
                return BasicSerialNo;
            } catch (Exception e) {
                log.error("解析序号配置异常", e);
                //解析异常，删除缓存
                clearCacheSerialNo(type);
                //重新初始化
                return initSerialNo(type, retryNum);
            }
        }
    }

    private BasicSerialNo checkRestGet(BasicSerialNo BasicSerialNo, String type, int retryNum) {
        SerialNoRuleEnum resetRule = SerialNoRuleEnum.getByName(BasicSerialNo.getSeqNoResetRule());
        if (resetRule != null) {
            int nowYear = LocalDate.now().getYear();
            int nowMonth = LocalDate.now().getMonthValue();
            int nowDay = LocalDate.now().getDayOfMonth();
            switch (resetRule) {
                case YEAR: {
                    //如果按年重置
                    if (BasicSerialNo.getLatestDate() == null || nowYear != BasicSerialNo.getLatestDate().getYear()) {
                        return resetSerialNo(type, retryNum);
                    }
                    break;
                }
                case MONTH: {
                    //如果是按月重置
                    if (BasicSerialNo.getLatestDate() == null || nowYear != BasicSerialNo.getLatestDate().getYear()
                            || nowMonth != BasicSerialNo.getLatestDate().getMonthValue()) {
                        return resetSerialNo(type, retryNum);
                    }
                    break;
                }
                case DAY: {
                    //如果是按天重置
                    if (BasicSerialNo.getLatestDate() == null || nowYear != BasicSerialNo.getLatestDate().getYear()
                            || nowMonth != BasicSerialNo.getLatestDate().getMonthValue()
                            || nowDay != BasicSerialNo.getLatestDate().getDayOfMonth()) {
                        return resetSerialNo(type, retryNum);
                    }
                    break;
                }
                case FOREVER: {
                    //不重置
                    break;
                }
                default:
                    //默认不重置
                    log.info("默认不重置");
                    break;
            }
        }
        return null;
    }

    private String getRedisKey(String type, RedisKey redisKey) {
        return String.join(":", FLAG_PREFIX, redisKey.name(), type);
    }
}
